package com.lvmama.java.rhino.etl.process;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.collections.CollectionUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.springframework.beans.BeanUtils;

import com.alibaba.fastjson.JSON;
import com.lvmama.java.rhino.etl.core.AbstractKafkaSparkStreamingTemplate;
import com.lvmama.java.rhino.spark.entity.AccessLog;
import com.lvmama.java.rhino.spark.entity.BaseLog;
import com.lvmama.java.rhino.spark.entity.DeviceInfoLog;
import com.lvmama.java.rhino.spark.entity.LogDetailInfo;
import com.lvmama.java.rhino.spark.entity.LogInfo;
import com.lvmama.java.rhino.spark.entity.RequestTimeInfo;
import com.lvmama.java.rhino.spark.entity.UserInfoLog;
import com.lvmama.java.rhino.spark.tables.RequestTimeDetail;
import com.lvmama.java.rhino.spark.tables.ElasticLogBaseParam;
import com.lvmama.java.rhino.spark.tables.ElasticRequestTimeDetail;
import com.lvmama.java.rhino.spark.tables.LogDetail;
import com.lvmama.java.rhino.spark.tables.RequestSpecialParam;
import com.lvmama.java.rhino.spark.tables.UserAccessPath;
import com.lvmama.java.rhino.spark.utils.Constants;
import com.lvmama.java.rhino.spark.utils.ElasticUtils;
import com.lvmama.java.rhino.spark.utils.LogComparator;
import com.lvmama.java.rhino.spark.utils.LogParseUtils;
import com.lvmama.java.rhino.spark.utils.LogProcessorUtil;
import com.lvmama.java.rhino.spark.utils.SaveToHdfsUtils;

public class SparkStreamingParseLogProcesser extends AbstractKafkaSparkStreamingTemplate {

	private static final long serialVersionUID = 9176498638269887343L;
	
//	private static final Logger LOGGER = Logger.getLogger(SparkStreamingParseLogProcesser.class);
	
	/** Map<接口名, Map<接口版本号, List<字段信息>>> 搜索特殊字段解析规则 */
//	private static final Map<String, Map<String, List<Map<String, String>>>> spacialSearchType = SpecialSearchTypeParseUtil.parse();;
	transient protected JavaSparkContext sc = null;
	transient protected SQLContext sqlContext = null;
	@Override
	public void excute(JavaDStream<String> lines) {
		// 创建Spark 上下文
		sc = new JavaSparkContext(sparkConf);
		// 创建Spark SQL上下文
		sqlContext = new SQLContext(sc);
		// 将一个请求的日志解析成对象
		JavaDStream<LogDetailInfo> tempRDD = lines.map(new Function<String, LogDetailInfo>() {
			private static final long serialVersionUID = 5260880512281497403L;
			@Override
			public LogDetailInfo call(String v1) throws Exception {
				List<String> logList = LogProcessorUtil.splitLog(v1);
				Collections.sort(logList, new LogComparator());
				return parseLog(logList);
			}
		});
		// 从日志对象中获取日志详细内容
		JavaDStream<LogDetail> logDetailRDD = tempRDD.map(new Function<LogDetailInfo, LogDetail>(){
			private static final long serialVersionUID = 6028449524795149897L;
			@Override
			public LogDetail call(LogDetailInfo v1) throws Exception {
				LogDetail logDetail = new LogDetail();
				logDetail.setThreadName(v1.getThreadName());
				logDetail.setContent(LogProcessorUtil.gzip(LogProcessorUtil.concatLog(v1.getLogList())));
				return logDetail;
			}
		});
		// 从日志对象中获取访问记录
		JavaDStream<UserAccessPath> userAccessPathRDD = tempRDD.map(new Function<LogDetailInfo, UserAccessPath>(){
			private static final long serialVersionUID = 6028449524795149897L;
			@Override
			public UserAccessPath call(LogDetailInfo v1) throws Exception {
				return combUserAccessPath(v1);
			}
		});
		// 获取请求时间详情
		JavaDStream<List<RequestTimeDetail>> requestTimeDetailRDD = tempRDD.map(new Function<LogDetailInfo, List<RequestTimeDetail>>(){
			private static final long serialVersionUID = 6028449524795149897L;
			@Override
			public List<RequestTimeDetail> call(LogDetailInfo v1) throws Exception {
				List<RequestTimeDetail> tempApiRequestTimeDetailList = v1.getApiRequestTime();
				return tempApiRequestTimeDetailList;
			}
		});
		// 获取应用的特殊字段
		JavaDStream<RequestSpecialParam> requestSpecialParamRDD = tempRDD.map(new Function<LogDetailInfo, RequestSpecialParam>(){
			private static final long serialVersionUID = 6028449524795149897L;
			@Override
			public RequestSpecialParam call(LogDetailInfo v1) throws Exception {
				return combRequestSpecialParam(v1);
			}
		});
		
		JavaDStream<ElasticLogBaseParam> elasticLogBaseParamRDD = tempRDD.map(new Function<LogDetailInfo, ElasticLogBaseParam>(){
			private static final long serialVersionUID = 6028449524795149897L;
			@Override
			public ElasticLogBaseParam call(LogDetailInfo v1) throws Exception {
				return combLogBaseParam(v1);
			}
		});
		
		logDetailRDD.foreachRDD(new VoidFunction<JavaRDD<LogDetail>>() {
			private static final long serialVersionUID = -5372210036478457103L;
			@Override
			public void call(JavaRDD<LogDetail> t) throws Exception {
				List<LogDetail> tempList = t.collect();
				if(tempList == null || tempList.isEmpty()) {
					return;
				}
				SaveToHdfsUtils.save(sqlContext, SaveToHdfsUtils.TABLE_NAME.LOG_DETAIL, tempList);	
			}
		});
		
		userAccessPathRDD.foreachRDD(new VoidFunction<JavaRDD<UserAccessPath>>() {
			private static final long serialVersionUID = -3457351055194183490L;
			@Override
			public void call(JavaRDD<UserAccessPath> t) throws Exception {
				List<UserAccessPath> tempList = t.collect();
				if(tempList == null || tempList.isEmpty()) {
					return;
				}
				SaveToHdfsUtils.save(sqlContext, SaveToHdfsUtils.TABLE_NAME.USER_ACCESS_PATH, tempList);
			}
		});
		
		requestTimeDetailRDD.foreachRDD(new VoidFunction<JavaRDD<List<RequestTimeDetail>>>() {
			private static final long serialVersionUID = -391057231573182644L;
			@Override
			public void call(JavaRDD<List<RequestTimeDetail>> t) throws Exception {
				List<List<RequestTimeDetail>> tempList = t.collect();
				if(tempList == null || tempList.isEmpty()) {
					return;
				}
				List<RequestTimeDetail> resultList = new ArrayList<RequestTimeDetail>();
				for(List<RequestTimeDetail> temp : tempList) {
					if(temp == null || temp.isEmpty()) {
						continue;
					}
					resultList.addAll(temp);
				}
				if(resultList == null || resultList.isEmpty()) {
					return;
				}
				SaveToHdfsUtils.save(sqlContext, SaveToHdfsUtils.TABLE_NAME.API_REQUEST_TIME_DETAIL, resultList);
			}
			
		});
		
		requestSpecialParamRDD.foreachRDD(new VoidFunction<JavaRDD<RequestSpecialParam>>() {
			private static final long serialVersionUID = 2908852184729646986L;
			@Override
			public void call(JavaRDD<RequestSpecialParam> t) throws Exception {
				List<RequestSpecialParam> tempList = t.collect();
				if(tempList == null || tempList.isEmpty()) {
					return;
				}
				List<RequestSpecialParam> resultList = new ArrayList<RequestSpecialParam>();
				for(RequestSpecialParam temp : tempList) {
					if(temp == null) {
						continue;
					}
					resultList.add(temp);
				}
				if(resultList.isEmpty()) {
					return;
				}
				SaveToHdfsUtils.save(sqlContext, SaveToHdfsUtils.TABLE_NAME.REQUEST_SPECIAL_PARAM, tempList);
			}
		});
		
		elasticLogBaseParamRDD.foreachRDD(new VoidFunction<JavaRDD<ElasticLogBaseParam>>() {
			private static final long serialVersionUID = 4674174838014369549L;
			@Override
			public void call(JavaRDD<ElasticLogBaseParam> t) throws Exception {
				List<ElasticLogBaseParam> tempList = t.collect();
				if(tempList == null || tempList.isEmpty()) {
					return;
				}
				for(ElasticLogBaseParam temp : tempList) {
					if(temp == null) {
						continue;
					}
					String index = Constants.ELASTIC_INDEX_LVTU_INDEX;
					String type = Constants.ELASTIC_TYPE_BASE_PARAM_TYPE;
					ElasticUtils.getInstance().putIndex(index, type, temp.getThreadName(), JSON.toJSONString(temp));
					System.out.println("Elasticsearch 存入完成。");
				}
			}
			
		});
		
	}
	

	private RequestSpecialParam combRequestSpecialParam(LogDetailInfo parseResult) {
		RequestSpecialParam requestSpecialParam = new RequestSpecialParam();
		List<String> specialParam = parseResult.getSpecialLog();
		if(specialParam == null) {
			return null;
		}
		for (int i = 0; i < specialParam.size(); i++) {
			if(i == 0) {
				requestSpecialParam.setParam1(specialParam.get(0));
			}
			if(i == 1) {
				requestSpecialParam.setParam2(specialParam.get(1));
			}
			if(i == 2) {
				requestSpecialParam.setParam3(specialParam.get(2));
			}
			if(i == 3) {
				requestSpecialParam.setParam4(specialParam.get(3));
			}
			if(i == 4) {
				requestSpecialParam.setParam5(specialParam.get(4));
			}
			if(i == 5) {
				requestSpecialParam.setParam6(specialParam.get(5));
			}
			if(i == 6) {
				requestSpecialParam.setParam7(specialParam.get(6));
			}
			if(i == 7) {
				requestSpecialParam.setParam8(specialParam.get(7));
			}
			if(i == 8) {
				requestSpecialParam.setParam9(specialParam.get(8));
			}
			if(i == 9) {
				requestSpecialParam.setParam10(specialParam.get(9));
			}
		}
		return requestSpecialParam;
	}

	private UserAccessPath combUserAccessPath(LogDetailInfo parseResult) {
		UserAccessPath userAccessPath = new UserAccessPath();
		userAccessPath.setThreadName(parseResult.getThreadName());
		userAccessPath.setAppName(parseResult.getAppName());
		BaseLog baseLog = parseResult.getBaseLog();
		if(baseLog != null) {
			if(baseLog.deviceInfo != null) {
				BeanUtils.copyProperties(baseLog.deviceInfo, userAccessPath);
			}
			if(baseLog.userInfo != null) {
				BeanUtils.copyProperties(baseLog.userInfo, userAccessPath);
			}
			if(baseLog.accessLog != null) {
				BeanUtils.copyProperties(baseLog.accessLog, userAccessPath);
			}
		}
		RequestTimeInfo requestTimeInfo = parseResult.getRequestTimeInfo();
		if(requestTimeInfo != null) {
			userAccessPath.setRequestTime(requestTimeInfo.getRequestTime());
			userAccessPath.setResponseTime(requestTimeInfo.getResponseTime());
		}
		return userAccessPath;
	}

	private ElasticLogBaseParam combLogBaseParam(LogDetailInfo parseResult) {
		ElasticLogBaseParam elasticLogBaseParam = new ElasticLogBaseParam();
		elasticLogBaseParam.setThreadName(parseResult.getThreadName());
		elasticLogBaseParam.setAppName(parseResult.getAppName());
		BaseLog baseLog = parseResult.getBaseLog();
		if(baseLog != null) {
			DeviceInfoLog deviceInfoLog = baseLog.deviceInfo;
			if(deviceInfoLog != null) {
				BeanUtils.copyProperties(deviceInfoLog, elasticLogBaseParam);
			}
			UserInfoLog userInfoLog = baseLog.userInfo;
			if(userInfoLog != null) {
				BeanUtils.copyProperties(userInfoLog, elasticLogBaseParam);
			}
			AccessLog accessLog = baseLog.accessLog;
			if(accessLog != null) {
				BeanUtils.copyProperties(accessLog, elasticLogBaseParam);
			}
		}
		RequestTimeInfo requestTimeInfo = parseResult.getRequestTimeInfo();
		if(requestTimeInfo != null) {
			BeanUtils.copyProperties(requestTimeInfo, elasticLogBaseParam);
		}
		elasticLogBaseParam.setLogDetail(parseResult.getLogList());
		List<String> specialLog = parseResult.getSpecialLog();
		if(specialLog != null) {
			if(specialLog.size() > 0) {
				elasticLogBaseParam.setParam1(specialLog.get(0));
			}
			if(specialLog.size() > 1) {
				elasticLogBaseParam.setParam2(specialLog.get(1));
			}
			if(specialLog.size() > 2) {
				elasticLogBaseParam.setParam3(specialLog.get(2));
			}
			if(specialLog.size() > 3) {
				elasticLogBaseParam.setParam4(specialLog.get(3));
			}
			if(specialLog.size() > 4) {
				elasticLogBaseParam.setParam5(specialLog.get(4));
			}
			if(specialLog.size() > 5) {
				elasticLogBaseParam.setParam6(specialLog.get(5));
			}
			if(specialLog.size() > 6) {
				elasticLogBaseParam.setParam7(specialLog.get(6));
			}
			if(specialLog.size() > 7) {
				elasticLogBaseParam.setParam8(specialLog.get(7));
			}
			if(specialLog.size() > 8) {
				elasticLogBaseParam.setParam9(specialLog.get(8));
			}
			if(specialLog.size() > 9) {
				elasticLogBaseParam.setParam10(specialLog.get(9));
			}
		}
		List<RequestTimeDetail> apiRequestTimeList = parseResult.getApiRequestTime();
		if(CollectionUtils.isNotEmpty(apiRequestTimeList)) {
			List<ElasticRequestTimeDetail> requestTimeDetail = new ArrayList<ElasticRequestTimeDetail>();
			for(RequestTimeDetail apiRequestTime : apiRequestTimeList) {
				ElasticRequestTimeDetail elasticRequestTimeDetail = new ElasticRequestTimeDetail();
				BeanUtils.copyProperties(apiRequestTime, elasticRequestTimeDetail);
				requestTimeDetail.add(elasticRequestTimeDetail);
			}
			elasticLogBaseParam.setRequestTimeDetail(requestTimeDetail);
		}
		return elasticLogBaseParam;
	}

	
	private LogDetailInfo parseLog(List<String> logList) {
		LogDetailInfo logDetailInfo = new LogDetailInfo();
		logDetailInfo.setLogList(logList);
		RequestTimeInfo requestTimeInfo = new RequestTimeInfo();
		logDetailInfo.setRequestTimeInfo(requestTimeInfo);
		for (int i = 0; i < logList.size(); i++) {
			LogInfo logInfo = LogParseUtils.parseLog(logList.get(i));
			if(logInfo == null) {
				continue;
			}
			if(Constants.COMMON_PARAM_DATA.equals(logInfo.getClassName())) {
				String content = logInfo.getContent();
				BaseLog baseLog = LogParseUtils.parseClientServiceCommonParam(content);
				logDetailInfo.setBaseLog(baseLog);
			}
			if(Constants.SPECIAL_PARAM_DATA.equals(logInfo.getClass())) {
				String content = logInfo.getContent();
				List<String> specialLog = LogParseUtils.parseClientServiceSpecialParam(content);
				logDetailInfo.setSpecialLog(specialLog);
			}
			if(i == 0) {
				Date printTime = logInfo.getPrintTime();
				requestTimeInfo.setRequestTime(printTime.getTime());
				logDetailInfo.setThreadName(logInfo.getThreadName());
				logDetailInfo.setAppName(logInfo.getAppName());
			}
			if(i == (logList.size()-1)) {
				Date printTime = logInfo.getPrintTime();
				requestTimeInfo.setResponseTime(printTime.getTime());
			}
		}
		logDetailInfo.setApiRequestTime(LogParseUtils.parseApiRequestTimeDetail(logList));
		return logDetailInfo;
	}
	
	@Override
	public Map<String, Integer> getKafkaTopic() {
		// Map<主题名, 线程数>
		Map<String, Integer> topicMap = new HashMap<String, Integer>();
		String topicsStr = Constants.getInstance().getValue("client.service.kafka.log.parse.topic");
		// 解析主题
		String[] topics = topicsStr.split(",");
		// 解析线程数
		int numThreads = Integer.parseInt(Constants.getInstance().getValue("client.service.spark.thread.nums"));
		for (String topic : topics) {
			topicMap.put(topic, numThreads);
		}
		return topicMap;
	}

	@Override
	public long getIntervalTime() {
		return 4000;
	}

	@Override
	public void saveEnvironment() {
		// TODO Auto-generated method stub

	}
}
